// Copyright 2013 The Go Circuit Project
// Use of this source code is governed by the license for
// The Go Circuit Project, found in the LICENSE file.
//
// Authors:
//   2013 Petar Maymounkov <p@gocircuit.org>

// Package kinfolk ...
package kinfolk

import (
	"bytes"
	"log"
	"sync"

	"circuit/kit/lang"
	"circuit/use/circuit"
)

// Kin
type Kin struct {
	permXID KinXID // Permanent circuit-wide unique ID for this kin
	rtr     *Rotor
	ach     chan KinXID // announcements
	dch     chan KinXID // denouncements
	sync.Mutex
	topic map[string]FolkXID // topic -> yfolk
	folk  []*Folk
}

// XKin
type XKin circuit.PermX

// FolkXID
type FolkXID XID

func (xid FolkXID) XID() XID {
	return XID(xid)
}

func (xid FolkXID) String() string {
	return "FolkXID." + XID(xid).String()
}

func NewKin(join XKin) (k *Kin, xkin ExoKin, add, rmv <-chan KinXID) {
	k = &Kin{
		rtr:   NewRotor(),
		ach:   make(chan KinXID, 3*Spread),
		dch:   make(chan KinXID, 3*Spread),
		topic: make(map[string]FolkXID),
	}
	k.permXID = KinXID(XID{
		X:  circuit.PermRef(ExoKin{k}),
		ID: lang.ComputeReceiverID(k),
	})
	if join == nil {
		//log.Println("peerless")
		return k, ExoKin{k}, k.ach, k.dch
	}
	var w bytes.Buffer
	//
	defer func() {
		if r := recover(); r != nil {
			log.Fatalf("joining (%s)", r)
		}
	}()
	for _, peer := range (YKin{KinXID{X: join}}).Join() {
		k.open(peer)
		w.WriteString(peer.X.Addr().WorkerID().String())
		w.WriteByte(' ')
	}
	//log.Println("peering", w.String())
	return k, ExoKin{k}, k.ach, k.dch
}

func (k *Kin) open(peer KinXID) KinXID {
	peer = KinXID(k.rtr.Open(XID(peer)))
	k.ach <- peer
	return peer
}

func (k *Kin) XID() KinXID {
	return k.permXID
}

func (k *Kin) replenish() {
	if k.rtr.NOpened() >= Spread {
		return
	}
	// Walk to a new ExoKin peer
	kinXID := ExoKin{k}.Walk(Depth)
	if XID(kinXID).IsNil() || kinXID.ID == k.XID().ID { // Compare just IDs, in case we got pointers to ourselves from elsewhere
		// If peer is nil or self, ignore it
		return
	}
	// Open the peer
	kinXID = k.open(kinXID)
	// Send new peer to all folk
	for _, folk := range k.snapfolk() {
		folk.supply(YKin{kinXID}.Attach(folk.Topic))
	}
}

func (k *Kin) snapfolk() []*Folk {
	k.Lock()
	defer k.Unlock()
	folk := make([]*Folk, len(k.folk))
	copy(folk, k.folk)
	return folk
}

func (k *Kin) scrub(kinXID KinXID) {
	//debug.PrintStack()
	if k.rtr.Scrub(kinXID.X).IsNil() {
		return
	}
	k.replenish()
	k.dch <- kinXID
}

func (k *Kin) Attach(topic string, folkXID FolkXID) *Folk {
	// Fetch initial peer connections
	var opened = k.rtr.Opened()
	peers := make([]FolkXID, 0, len(opened))
	for _, xid := range opened {
		kinXID := KinXID(xid)
		if folkXID := (YKin{kinXID}).Attach(topic); folkXID.X != nil {
			// Scrub kinXID if folkXID.X ever panics
			folkXID.X = watch(folkXID.X, func(_ circuit.PermX, r interface{}) {
				k.scrub(kinXID)
				panic(r)
			})
			peers = append(peers, folkXID)
		}
	}

	k.Lock()
	defer k.Unlock()

	if _, present := k.topic[topic]; present {
		panic("dup attach")
	}
	folk := &Folk{
		Topic: topic,
		rtr:   NewRotor(),
		ch:    make(chan FolkXID, 6*(1+len(peers))),
	}
	// Inject initial peers
	for _, peer := range peers {
		folk.supply(peer)
	}

	// Register
	k.folk = append(k.folk, folk)
	k.topic[topic] = folkXID

	return folk
}

// KinXID
type KinXID XID

func (xid KinXID) String() string {
	return "KinXID." + XID(xid).String()
}

// ExoKin
type ExoKin struct {
	k *Kin
}

// Attach returns a cross-reference to an exo folk object
func (x ExoKin) Attach(topic string) FolkXID {
	x.k.Lock()
	defer x.k.Unlock()
	return x.k.topic[topic]
}

// Join returns an initial set of peers that the joining kin should use as initial entry into the kin system.
func (x ExoKin) Join() []KinXID {
	m := make(map[lang.ReceiverID]KinXID)
	for i := 0; i < Spread; i++ {
		peerXID := x.Walk(Depth)
		if XID(peerXID).IsNil() {
			continue
		}
		if _, ok := m[peerXID.ID]; ok {
			// Duplicate
			continue
		}
		m[peerXID.ID] = peerXID
	}
	r := make([]KinXID, 0, len(m))
	for _, peerXID := range m {
		r = append(r, peerXID)
	}
	return r
}

func (x ExoKin) Walk(t int) KinXID {
	if t <= 0 {
		return x.k.XID()
	}
	hop := KinXID(x.k.rtr.Choose())
	if hop.X == nil {
		return x.k.XID()
	}
	defer func() {
		if r := recover(); r != nil {
			x.k.scrub(hop)
		}
	}()
	return YKin{hop}.Walk(t - 1)
}

// YKin
type YKin struct {
	xid KinXID
}

func (y YKin) Join() []KinXID {
	// Do not recover
	return y.xid.X.Call("Join")[0].([]KinXID)
}

func (y YKin) Walk(t int) KinXID {
	// Do not recover; ExoKin.Walk relies on panics
	return y.xid.X.Call("Walk", t)[0].(KinXID)
}

func (y YKin) Attach(topic string) FolkXID {
	defer func() {
		recover()
	}()
	return y.xid.X.Call("Attach", topic)[0].(FolkXID)
}

// Init
func init() {
	circuit.RegisterValue(&Kin{}) // So that we can computer receiver ID
	circuit.RegisterValue(ExoKin{})
}

const (
	Spread = 5
	Depth  = 5
)
